package com.lc1993929.mr.flowlog.enhance;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;

/**
 * Created by LiuChang on 2018/1/9.
 */
public class FlowLogEnhance2 {

    static class FlowLogEnhance2Mapper extends Mapper<LongWritable, Text, Text, NullWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] words = line.split("");
            for (String word : words) {
                context.write(new Text(word), NullWritable.get());
            }
        }
    }

    static class FlowLogEnhance2Reducer extends Reducer<Text, NullWritable, Text, NullWritable> {
        MultipleOutputs<Text, NullWritable> multipleOutputs;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            multipleOutputs = new MultipleOutputs<>(context);
        }

        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            if (key.toString().startsWith("a")) {
                multipleOutputs.write("A", key, NullWritable.get(), "a/");
            } else if (key.toString().startsWith("b")) {
                multipleOutputs.write("B", key, NullWritable.get(), "b/");
            } else {
                multipleOutputs.write("C", key, NullWritable.get(), "other/");
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            multipleOutputs.close();
        }
    }

    public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {
        Configuration conf = new Configuration();
        //conf.set("mapreduce.framework.name", "local");
        Job job = Job.getInstance(conf);
        //要执行的jar文件
        job.setJarByClass(FlowLogEnhance2.class);
        //指定mapper类和reducer类
        job.setMapperClass(FlowLogEnhance2Mapper.class);
        job.setReducerClass(FlowLogEnhance2Reducer.class);
        //指定mapper和reducer的输出的key和value的数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);
        //指定输出规则
        job.setOutputFormatClass(EnhanceOutputFormat.class);
        //指定要解析的文件路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //指定输出的文件路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //不生成part-r0000文件
        LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
        //定义输出文件别名
        MultipleOutputs.addNamedOutput(job, "A", TextOutputFormat.class, Text.class, NullWritable.class);
        MultipleOutputs.addNamedOutput(job, "B", TextOutputFormat.class, Text.class, NullWritable.class);
        MultipleOutputs.addNamedOutput(job, "C", TextOutputFormat.class, Text.class, NullWritable.class);

        job.setNumReduceTasks(1);
        // 向yarn提交job
        boolean res = job.waitForCompletion(true);

        System.exit(res ? 0 : 1);
    }
}
